package cn.gwm.flink.streaming.function.apply;

import cn.gwm.flink.streaming.beans.BeanSource;
import cn.gwm.flink.streaming.constant.BaseFields;
import cn.hutool.json.JSONObject;
import org.apache.flink.api.common.functions.CoGroupFunction;
import org.apache.flink.api.common.functions.FlatJoinFunction;
import org.apache.flink.util.Collector;

import java.util.Iterator;

/**
 * @ClassName CanGpsCoGroupFunction
 * @Author xzh
 * @Date 2022/11/16 13:21
 **/
public class CanGpsCoGroupFunction  implements CoGroupFunction<JSONObject,JSONObject,JSONObject> {

    BeanSource.SourceEnum sourceEnum;

    public CanGpsCoGroupFunction(BeanSource.SourceEnum sourceEnum){
        this.sourceEnum = sourceEnum;
    }

    @Override
    public void coGroup(Iterable<JSONObject> first, Iterable<JSONObject> second, Collector<JSONObject> out) throws Exception {
        Iterator<JSONObject> can = first.iterator();
        while(can.hasNext()) {
            JSONObject canObj = can.next();
            Iterator<JSONObject> gps = second.iterator();
            while(gps.hasNext()) {
                JSONObject gpsObj = gps.next();
                if (gpsObj != null && gpsObj.getLong(BaseFields.tid).longValue() == canObj.getLong(BaseFields.tid).longValue()) {
                    canObj.set("tboxMoniGPSstatus", gpsObj.getStr("tboxMoniGPSstatus"));
                    canObj.set("tboxMoniLongitude", gpsObj.getStr("tboxMoniLongitude"));
                    canObj.set("tboxMoniLatitude", gpsObj.getStr("tboxMoniLatitude"));
                    canObj.set("imsi2", gpsObj.getStr("imsi"));
                    canObj.set("command2", gpsObj.getStr("command"));
                    canObj.set("user_id2", gpsObj.getStr("user_id"));
                }
            }

            out.collect(BeanSource.transJson(canObj,sourceEnum));
        }
    }
}
